package tasktracker.abstraction.extensions;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import tasktracker.abstraction.AbstractTaskTracker;
import tasktracker.abstraction.TaskInfo;
import tasktracker.abstraction.TaskResultInfo;
import tasktracker.abstraction.TaskStatus;

public abstract class StandaloneProcessTaskTracker extends AbstractTaskTracker {

	private Map<String, TaskInfo> taskInfos = new ConcurrentHashMap<String, TaskInfo>();
	private Map<String, TaskStatusTracker> statusTrackers = new HashMap<String, TaskStatusTracker>();
	private Map<String, Process> taskProcesses = new ConcurrentHashMap<String, Process>();
	private int checkInterval;

	protected StandaloneProcessTaskTracker(int checkInterval) {
		this.checkInterval = checkInterval;
	}

	private boolean shuttingDown;
	private Object shuttedDown;

	@Override
	public void init() throws Exception {
		shuttingDown = false;
		shuttedDown = new Object();
		Thread thread = new Thread(new Runnable() {
			@Override
			public void run() {
				watcherThread();
			}
		});
		thread.start();
	}

	@Override
	public void destroy() {
		shuttingDown = true;
		synchronized (shuttedDown) {
			try {
				shuttedDown.wait();
			} catch (InterruptedException e) {
				logger.warn("StandaloneProcessTaskTracker.destroy()", e);
			}

		}
	}

	private void watcherThread() {
		while (!shuttingDown) {
			try {
				Thread.sleep(checkInterval);
			} catch (InterruptedException e) {
				logger.warn("StandaloneProcessTaskTracker.watcherThread()", e);
				break;
			}

			for (Iterator<Map.Entry<String, Process>> i = taskProcesses.entrySet().iterator(); i.hasNext();) {
				Map.Entry<String, Process> entry = i.next();

				Process process = entry.getValue();
				if (!process.isAlive()) {
					String guid = entry.getKey();
					processResult(guid, process);
					TaskStatusTracker statusTracker = statusTrackers.get(guid);
					if (statusTracker != null && !statusTracker.isCompleted()) {
						int exitCode = process.exitValue();
						if (exitCode == 0) {
							reportStatus(guid, TaskStatus.COMPLETED, null);
						} else {
							reportStatus(guid, TaskStatus.ERROR, null);
						}
					}
					i.remove();
				}
			}
		}
		shuttedDown.notify();
	}

	protected abstract ProcessBuilder getProcessBuilder(TaskInfo task);

	protected abstract void processResult(String guid, Process process);

	protected TaskInfo getTaskInfo(String guid) {
		return taskInfos.get(guid);
	}

	@Override
	public void startTask(TaskInfo task) throws Exception {
		String guid = task.getGuid();
		taskInfos.put(task.getGuid(), task);
		TaskStatusTracker statusTracker = new TaskStatusTracker();
		statusTrackers.put(guid, statusTracker);

		ProcessBuilder pb = null;
		try {
			pb = getProcessBuilder(task);
		} catch (Exception e) {
			reportStatus(guid, TaskStatus.ERROR, new TaskResultInfo(e.toString()));
			return;
		}

		if (pb == null) {
			if (!statusTracker.isCompleted())
				reportStatus(guid, TaskStatus.ERROR, new TaskResultInfo("No process info provided."));
			return;
		}

		Process process = pb.start();
		taskProcesses.put(task.getGuid(), process);
	}

	@Override
	protected void reportStatus(String guid, int status, TaskResultInfo result) {
		TaskStatusTracker statusTracker = statusTrackers.get(guid);
		if (statusTracker == null)
			return;

		statusTracker.update(status);
		super.reportStatus(guid, status, result);
	}

	@Override
	public boolean cleanupTask(String guid) throws Exception {
		taskInfos.remove(guid);
		statusTrackers.remove(guid);
		taskProcesses.remove(guid);
		return false;
	}

}
